package cn.doitedu.demo2;

import cn.doitedu.demo1.EventBean;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.apache.http.HttpHost;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.*;
import org.elasticsearch.search.builder.SearchSourceBuilder;

import java.io.IOException;

/**
 * -- 营销策略2：营销规则
 * 针对月平均消费额 > 300， 性别为 male的用户，                     -- 静态画像条件
 * 当他发生 3次 以上 点赞 行为后                                  -- 实时动态画像条件
 * 等他再发生 下单行为，即给他推送一个 xxxx 优惠活动信息           -- 触发条件
 */
public class RuleCalculatorV2_Rule2 implements RuleCalculatorV2 {

    RestHighLevelClient client;
    SearchRequest request;
    JSONObject message;
    ValueState<Integer> cntState;


    @Override
    public void init(RuntimeContext runtimeContext) {
        // es的请求客户端
        client = new RestHighLevelClient(
                RestClient
                        .builder(new HttpHost("doitedu", 9200, "http"))
        );
        // 用于查询参数封装的对象
        request = new SearchRequest("doit42_userprofile");

        message = new JSONObject();
        message.put("rule_id", "rule-002");


        // 获取一个用于记录点赞次数的valueState
        cntState = runtimeContext.getState(new ValueStateDescriptor<Integer>("rule-002-state", Integer.class));

    }

    @Override
    public void calculate(EventBean eventBean, Collector<String> out) throws IOException {

        String eventId = eventBean.getEvent_id();
        Long userId = eventBean.getUser_id();

        // 当前事件是否是动态画像要统计的事件
        // 如果是，则进行画像统计
        if ("点赞".equals(eventId)) {
            cntState.update(cntState.value() == null ? 1 : cntState.value() + 1);
        }

        // 当前事件是否是触发条件要求的事件
        // 如果是，则判断该用户的：动态画像条件和静态画像条件是否满足，如满足，则推送消息
        if ("下单".equals(eventId)) {

            // 判断动态画像条件是否满足
            if (cntState.value() != null && cntState.value() >= 3) {

                // 去查询该用户的静态画像条件是否满足
                // 针对月平均消费额 > 220， 性别为 male的用户
                IdsQueryBuilder idsQueryBuilder = QueryBuilders.idsQuery().addIds(userId + "");
                RangeQueryBuilder tag0101 = QueryBuilders.rangeQuery("tag0101").gt(200);
                MatchQueryBuilder tag0202 = QueryBuilders.matchQuery("tag0201", "male");
                BoolQueryBuilder must = QueryBuilders.boolQuery().must(idsQueryBuilder).must(tag0101).must(tag0202);

                request.source(new SearchSourceBuilder().query(must));
                SearchResponse response = client.search(request, RequestOptions.DEFAULT);
                if (response.getHits().getTotalHits().value > 0) {
                    // 走到这里，说明静态画像条件满足
                    message.put("user_id", userId);
                    message.put("hit_time", eventBean.getAction_time());

                    out.collect(message.toJSONString());

                }


            }


        }


    }
}
